package org.example.controller;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.example.model.AppOrder;
import org.example.model.CustomInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;

@RestController
@RequestMapping("/push")
@Slf4j
public class PushController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping(value = "/pushAppOrder",method = RequestMethod.POST)
    public String pushAppOrder(@RequestBody AppOrder appOrder){
        //发送带有时间戳的消息
        kafkaTemplate.send("hello", 0, System.currentTimeMillis(), "appOrder", JSON.toJSONString(appOrder));

        //使用ProducerRecord发送消息
        ProducerRecord record = new ProducerRecord("hello", "appOrder", JSON.toJSONString(appOrder));
        kafkaTemplate.send(record);

        //使用Message发送消息
        Map map = new HashMap();
        map.put(KafkaHeaders.TOPIC, "hello");
        map.put(KafkaHeaders.PARTITION_ID, 0);
        map.put(KafkaHeaders.MESSAGE_KEY, "appOrder");
        GenericMessage message = new GenericMessage(JSON.toJSONString(appOrder),new MessageHeaders(map));
        kafkaTemplate.send(message);
        return "pushAppOrder Success";
    }

    @RequestMapping(value = "/pushCustomInfo",method = RequestMethod.POST)
    public String pushCustomInfo(@RequestBody CustomInfo customInfo){
        //发送带有时间戳的消息
        kafkaTemplate.send("hello", 0, System.currentTimeMillis(), "customInfo", JSON.toJSONString(customInfo));

        //使用ProducerRecord发送消息
        ProducerRecord record = new ProducerRecord("hello", "customInfo", JSON.toJSONString(customInfo));
        kafkaTemplate.send(record);

        //使用Message发送消息
        Map map = new HashMap();
        map.put(KafkaHeaders.TOPIC, "hello");
        map.put(KafkaHeaders.PARTITION_ID, 0);
        map.put(KafkaHeaders.MESSAGE_KEY, "customInfo");
        GenericMessage message = new GenericMessage(JSON.toJSONString(customInfo),new MessageHeaders(map));
        kafkaTemplate.send(message);
        return "pushCustomInfo Success";
    }

}
